package com.ls.mq.listener.impl;

import org.apache.log4j.Logger;

import com.ls.mq.actor.Actor;
import com.ls.mq.actor.impl.ActorRefImpl;
import com.ls.mq.iterface.IMQFuture;
import com.ls.mq.iterface.MessageListener;
import com.ls.mq.manager.FutureManager;
import com.ls.mq.message.Message;
import com.ls.mq.thread.ExecutorsManager;
import com.ls.mq.util.Toolkit;

public class DefaultMessageListener implements MessageListener {
	Logger logger = Logger.getLogger(DefaultMessageListener.class);

	@Override
	public void doMessage(String channel, String msg) {
		ExecutorsManager.getInstance().submit(new ActorMessageHandler(channel, msg));// 每个线程一个单独的类。每个消息一个单独的类，那么可以认为方法是线程安全的
	}
}

class ActorMessageHandler implements Runnable {
	private String channel;
	private String msg;

	public ActorMessageHandler(String channel, String msg) {
		super();
		this.channel = channel;
		this.msg = msg;
	}

	@SuppressWarnings({ "rawtypes" })
	public void process(String channel, String msg) {
		Message message = Toolkit.fromJson(msg, Message.class);
		if (Toolkit.isEmpty(message))
			return;
		if (message.isRequest()) {
			Object obj = message.getDataObj();
			String actor = message.getDest().getActor();
			try {
				Class actorClz = Class.forName(actor);
				Actor actorObj = (Actor) actorClz.newInstance();
				Object result = null;
				try {
					result = actorObj.doMessage(obj);
				} catch (Throwable e) {
					result = e;
				}
				if (message.isRequestReply()) {
					ActorRefImpl ref = ActorRefImpl.build(message);
					ref.replyUnsafe(result);
					System.out.println("reply");
				}
				System.out.println("task finished with result : " + result);
			} catch (ClassNotFoundException | InstantiationException | IllegalAccessException e) {
				e.printStackTrace();
			}
		} else {
			System.out.println("in request reply");
			String uuid = message.getUuid();
			Object reply = message.getDataObj();
			IMQFuture future = FutureManager.getInstance().getFuture(uuid);
			if (future != null)
				future.set(reply);
		}
	}

	@Override
	public void run() {
		process(getChannel(), getMsg());
	}

	public String getMsg() {
		return msg;
	}

	public void setMsg(String msg) {
		this.msg = msg;
	}

	public String getChannel() {
		return channel;
	}

	public void setChannel(String channel) {
		this.channel = channel;
	}
}